package com.wg.neptune.zookeeper;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.wg.neptune.help.IPHelper;
import com.wg.neptune.help.PropertyConfigHelper;
import com.wg.neptune.model.ConsumerService;
import com.wg.neptune.model.ProviderService;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.SerializableSerializer;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.ClassUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * Created by mc on 18/5/31.
 */
public class RegisterCenter implements IProviderRegister ,IConsumerRegister{

    private static final Logger logger = LoggerFactory.getLogger(RegisterCenter.class);

    private static RegisterCenter registerCenter = new RegisterCenter();

    /**
     * 服务提供者列表,Key:服务提供者接口  value:服务提供者服务方法列表
     * 用于提供保存服务列表的初始化
     */
    private static final Map<String, List<ProviderService>> providerServiceMap = Maps.newConcurrentMap();

    /**
     * 服务端ZK服务元信息,选择服务(第一次直接从ZK拉取,后续由ZK的监听机制主动更新)
     */
    private static final Map<String, List<ProviderService>> serviceMetaDataMap4Consume = com.google.common.collect.Maps.newConcurrentMap();

    private static String ZK_SERVICE = PropertyConfigHelper.getZkService();
    private static int ZK_SESSION_TIME_OUT = PropertyConfigHelper.getZkConnectionTimeout();
    private static int ZK_CONNECTION_TIME_OUT = PropertyConfigHelper.getZkConnectionTimeout();
    private static String ROOT_PATH = "/neptune-rpc";
    public static String PROVIDER_TYPE = "provider";
    public static String INVOKER_TYPE = "consumer";
    private static volatile ZkClient zkClient = null;


    private RegisterCenter() {
    }
    /**
     * 单例模式
     * @return
     */
    public static RegisterCenter singleton() {
        return registerCenter;
    }

    /**
     * 注册服务至zk
     * @param providerServices
     */
    public void providerRegister(final List<ProviderService> providerServices) {
        //判空
        if(CollectionUtils.isEmpty(providerServices)){
            logger.error("服务提供列表不能为空");
            return;
        }
        //查看providerServiceMap 是否存在ProviderService，如果不存在注册一把
        for(ProviderService providerService : providerServices){
            String serviceItfKey = providerService.getServiceItf().getName();
            List<ProviderService> providers = providerServiceMap.get(providerService);
            if (providers == null) {
                providers = Lists.newArrayList();
            }
            providers.add(providerService);
            providerServiceMap.put(serviceItfKey, providers);
        }

        if (zkClient == null) {
            zkClient = new ZkClient("localhost:2181", 1000, 1000, new SerializableSerializer());
        }
        //创建 ZK命名空间/当前部署应用APP命名空间/  一个app只有一个命名空间
        String APP_KEY = providerServices.get(0).getAppKey();
        String ZK_PATH = ROOT_PATH + "/" + APP_KEY;
        boolean exist = zkClient.exists(ZK_PATH);
        if (!exist) {
            zkClient.createPersistent(ZK_PATH, true);
        }
        for (Map.Entry<String, List<ProviderService>> entry : providerServiceMap.entrySet()) {
            //服务分组
            String groupName = entry.getValue().get(0).getGroupName();
            /**
             * 创建服务提供者
             * 以接口为单位：不同的服务器有不同的ip，是一个服务器接口列表
             */
            String serviceNode = entry.getKey();
            String servicePath = ZK_PATH + "/" + groupName + "/" + serviceNode + "/" + PROVIDER_TYPE;
            exist = zkClient.exists(servicePath);
            if (!exist) {
                zkClient.createPersistent(servicePath, true);
            }

            /**
             *  创建当前服务器节点
             *  跟ip注册接口的一个服务提供者接口，可以有多个ip
             */
            int serverPort = entry.getValue().get(0).getServerPort();//服务端口
            int weight = entry.getValue().get(0).getWeight();//服务权重
            int workerThreads = entry.getValue().get(0).getWorkThreads();//服务工作线程
            String localIp = IPHelper.localIp();
            String currentServiceIpNode = servicePath + "/" + localIp + "," + serverPort + "," + weight + "," + workerThreads + "," + groupName;
            exist = zkClient.exists(currentServiceIpNode);
            if (!exist) {
                //注意,这里创建的是临时节点
                zkClient.createEphemeral(currentServiceIpNode);
            }

            /**
             * 监听服务的动态变化，将不存在的服务直接从本地缓存中干掉，把ip当成key
             */
            zkClient.subscribeChildChanges(servicePath, new IZkChildListener() {
                public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                    if (currentChilds == null) {
                        currentChilds = Lists.newArrayList();
                    }

                    //存活的服务IP列表
                    List<String> activityServiceIpList = Lists.newArrayList(Lists.transform(currentChilds, new Function<String, String>() {
                        public String apply(String input) {
                            return StringUtils.split(input, ",")[0];
                        }
                    }));
                    refreshActivityService(activityServiceIpList);
                }
            });
        }
    }

    /**
     * 利用ZK自动刷新当前存活的服务提供者列表数据
     */
    private void refreshActivityService(List<String> serviceIpList) {
        if (serviceIpList == null) {
            serviceIpList = Lists.newArrayList();
        }

        Map<String, List<ProviderService>> currentServiceMetaDataMap = Maps.newHashMap();
        for (Map.Entry<String, List<ProviderService>> entry : providerServiceMap.entrySet()) {
            String key = entry.getKey();
            List<ProviderService> providerServices = entry.getValue();

            List<ProviderService> serviceMetaDataModelList = currentServiceMetaDataMap.get(key);
            if (serviceMetaDataModelList == null) {
                serviceMetaDataModelList = Lists.newArrayList();
            }

            for (ProviderService serviceMetaData : providerServices) {
                if (serviceIpList.contains(serviceMetaData.getServerIp())) {
                    serviceMetaDataModelList.add(serviceMetaData);
                }
            }
            currentServiceMetaDataMap.put(key, serviceMetaDataModelList);
        }
        providerServiceMap.clear();
        System.out.println("currentServiceMetaDataMap,"+ JSON.toJSONString(currentServiceMetaDataMap));
        providerServiceMap.putAll(currentServiceMetaDataMap);
    }



    /**
     * 获取zk服务列表
     * @return
     */
    public Map<String, List<ProviderService>> getProviderServiceMap() {
        return providerServiceMap;
    }



    /*******************************************************************************************************************************/
    /**
     * 初始化服务提供者信息
     */
    public void initProviderMap(String appKey, String groupName) {
        if (MapUtils.isEmpty(serviceMetaDataMap4Consume)) {
            serviceMetaDataMap4Consume.putAll(fetchOrUpdateServiceMetaData(appKey, groupName));
        }
    }

    private Map<String, List<ProviderService>> fetchOrUpdateServiceMetaData(String appKey, String groupName) {
        final Map<String, List<ProviderService>> providerServiceMap = Maps.newConcurrentMap();
        //连接zk
        synchronized (RegisterCenter.class) {
            if (zkClient == null) {
                zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIME_OUT, new SerializableSerializer());
            }
        }

        //从ZK获取服务提供者列表
        String providePath = ROOT_PATH + "/" + appKey + "/" + groupName;
        List<String> providerServices = zkClient.getChildren(providePath);

        for (String serviceName : providerServices) {
            String servicePath = providePath + "/" + serviceName + "/" + PROVIDER_TYPE;
            List<String> ipPathList = zkClient.getChildren(servicePath);
            for (String ipPath : ipPathList) {
                String serverIp = StringUtils.split(ipPath, ",")[0];
                String serverPort = StringUtils.split(ipPath, ",")[1];
                int weight = Integer.parseInt(StringUtils.split(ipPath, ",")[2]);
                int workerThreads = Integer.parseInt(StringUtils.split(ipPath, ",")[3]);
                String group = StringUtils.split(ipPath, ",")[4];

                List<ProviderService> providerServiceList = providerServiceMap.get(serviceName);
                if (providerServiceList == null) {
                    providerServiceList = Lists.newArrayList();
                }
                ProviderService providerService = new ProviderService();

                try {
                    providerService.setServiceItf(ClassUtils.getClass(serviceName));
                } catch (ClassNotFoundException e) {
                    throw new RuntimeException(e);
                }

                providerService.setServerIp(serverIp);
                providerService.setServerPort(Integer.parseInt(serverPort));
                providerService.setWeight(weight);
                providerService.setWorkThreads(workerThreads);
                providerService.setGroupName(group);
                providerServiceList.add(providerService);

                providerServiceMap.put(serviceName, providerServiceList);
            }

            //监听注册服务的变化,同时更新数据到本地缓存
            zkClient.subscribeChildChanges(servicePath, new IZkChildListener() {
                public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                    if (currentChilds == null) {
                        currentChilds = Lists.newArrayList();
                    }
                    currentChilds = Lists.newArrayList(Lists.transform(currentChilds, new Function<String, String>() {
                        public String apply(String input) {
                            return StringUtils.split(input, "|")[0];
                        }
                    }));
                    refreshServiceMetaDataMap(currentChilds);
                }
            });
        }
        return providerServiceMap;
    }

    private void refreshServiceMetaDataMap(List<String> serviceIpList) {
        if (serviceIpList == null) {
            serviceIpList = Lists.newArrayList();
        }
        Map<String, List<ProviderService>> currentServiceMetaDataMap = Maps.newHashMap();
        for (Map.Entry<String, List<ProviderService>> entry : serviceMetaDataMap4Consume.entrySet()) {
            String serviceItfKey = entry.getKey();
            List<ProviderService> serviceList = entry.getValue();

            List<ProviderService> providerServiceList = currentServiceMetaDataMap.get(serviceItfKey);
            if (providerServiceList == null) {
                providerServiceList = Lists.newArrayList();
            }

            for (ProviderService serviceMetaData : serviceList) {
                if (serviceIpList.contains(serviceMetaData.getServerIp())) {
                    providerServiceList.add(serviceMetaData);
                }
            }
            currentServiceMetaDataMap.put(serviceItfKey, providerServiceList);
        }

        serviceMetaDataMap4Consume.clear();
        serviceMetaDataMap4Consume.putAll(currentServiceMetaDataMap);
    }
    /**
     * 给消费者初始化服务提供者map
     * @return
     */
    public Map<String, List<ProviderService>> getServiceMap4Consumer() {
        return serviceMetaDataMap4Consume;
    }

    /**
     * 注册本地消费者信息
     * @param consumerService
     */
    public void registerConsemer(ConsumerService consumerService) {
        if (consumerService == null) {
            return;
        }

        //连接zk,注册服务
        synchronized (RegisterCenter.class) {

            if (zkClient == null) {
                zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIME_OUT, new SerializableSerializer());
            }
            //创建 ZK命名空间/当前部署应用APP命名空间/
            boolean exist = zkClient.exists(ROOT_PATH);
            if (!exist) {
                zkClient.createPersistent(ROOT_PATH, true);
            }


            //创建服务消费者节点
            String appKey = consumerService.getAppKey();
            String groupName = consumerService.getGroupName();
            String serviceNode = consumerService.getServiceItf().getName();
            String servicePath = ROOT_PATH + "/" + appKey + "/" + groupName + "/" + serviceNode + "/" + INVOKER_TYPE;
            exist = zkClient.exists(servicePath);
            if (!exist) {
                zkClient.createPersistent(servicePath, true);
            }

            //创建当前服务器节点
            String localIp = IPHelper.localIp();
            String currentServiceIpNode = servicePath + "/" + localIp;
            exist = zkClient.exists(currentServiceIpNode);
            if (!exist) {
                //注意,这里创建的是临时节点
                zkClient.createEphemeral(currentServiceIpNode);
            }
        }
    }

    /**
     * 测试服务注册至zk，本地测试
     * @param args
     */
    public static void main(String[] args) {
        List<ProviderService> providerServices = Lists.newArrayList();
        ProviderService providerService = new ProviderService();
        Date date = new Date();
        providerService.setAppKey("rpc");
        providerService.setServerPort(8088);
        providerService.setServiceObject(date);
        providerService.setServiceItf(Date.class);
        providerServices.add(providerService);
        registerCenter.providerRegister(providerServices);
    }
}
